use std::borrow::Borrow;
use std::collections::HashMap;
use std::{rc::Rc, ops::Deref};
use std::sync::{Arc, Mutex};

use tokio::sync::mpsc::{Receiver, Sender};
use tokio_stream::wrappers::ReceiverStream;

use crate::logger::logger::Logger;

use super::{storage::StorageSystem, message::Message};

/**
 * @author Almas software technologies muhtarjaan mahmood (مۇختەرجان مەخمۇت)
 * @email ug-project@outlook.com
 * @create date 2022-07-05 18:48:54
 * @modify date 2022-07-05 18:48:54
 * @desc [description]
*/

/// Indicates importance of message in the message pipe,
/// The importance of Some messages is very critical, we must be ensure the message is fully written to the local file system before precessing it, but some kind of messages are often not very critical so we can schedule the store task to after few seconds. this is can much more speed up write performance
#[derive(Clone)]
pub enum StoringImportance {
    /// Messages that incoming to the message pipe is write to the local file system before out if importance is Critical,
    Critical,
    /// Messages that incoming to the message pipe is directly out before message is stored to the local file system, the store task is scheduled after few seconds for performance considerations.
    Usual,
}

pub struct MessagePipe<LOGGER: Logger + Send + Sync> {
    pub uuid: String,
    pub name: String,
    pub description: String,
    pub message_importance: StoringImportance,
    pub has_table: bool,
    pub storage: Arc<dyn StorageSystem + Send + Sync>,
    ///Inputs message from TransformerApplication
    pub input: Option<Sender<Message>>,
    ///Outputs message to TransformerApplication
    pub output: Arc<Mutex<HashMap<String, Sender<Message>>>>,
    pub logger: Arc<LOGGER>,
}


impl<LOGGER: Logger + Send + Sync + 'static> MessagePipe<LOGGER> {
    pub fn new(id: String, name: String, description: String, importance: StoringImportance, has_table: bool, storage_system: Arc<dyn StorageSystem + Send + Sync>, logger: Arc<LOGGER>) -> MessagePipe<LOGGER> {
        let mut message = MessagePipe {
            name,
            message_importance: importance,
            has_table,
            uuid: id,
            storage: storage_system,
            input: None,
            output: Arc::new(Mutex::new(HashMap::new())),
            logger,
            description
        };
        message.initialize();
        message
    }
    fn initialize(&mut self) {
        let (sender, mut receiver) = tokio::sync::mpsc::channel(8192);
        self.input = Some(sender);
        let outputs = self.output.clone();
        {
            let (
                storage,
                importance_cloned,
                name,
                logger
            ) = (
                self.storage.clone(),
                self.message_importance.clone(),
                self.name.clone(),
                self.logger.clone()
            );
            let outputs = self.output.clone();
            tokio::spawn(async move {
                logger.deref().ok(format!("Start of core processing loop of Pipe: '{}'", name));
                while let Some(message) = receiver.recv().await {
                    match importance_cloned {
                        StoringImportance::Critical => {
                            (*storage).store_immediately(&message).await;
                        },        
                        StoringImportance::Usual => {
                            (*storage).store_with_schedule(&message).await;
                        },
                    }
                    let outputs = {
                        Vec::from_iter(outputs.lock().expect("Cannot acquire pipe outputs").iter().map(|port|port.1.clone()))
                    };
                    for port in outputs {
                        port.send(message.clone()).await.expect("Cannot write message to app");
                    };
                }
                logger.deref().info(format!("MessagePipe core processing loop is OVER."));
            });
        }
    }
    // pub async fn write_message(&self, message: &Message) {
    //     match self.message_importance {
    //         StoringImportance::Critical => self.storage.store_immediately(message).await,
    //         StoringImportance::Usual => self.storage.store_with_schedule(message).await,
    //     };
    // }
}
